extern "C" {
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/time.h>
}
#include <iostream>
#include <map>
#include <queue>
#include <string>
#include <vector>

#include "http.hpp"
#include "logger.hpp"

volatile sig_atomic_t g_stop_server = false;

class Server {
private:
  const int port;
  const int worker_num;
  std::vector<pthread_t> workers;
  const std::string doc_root;

  std::queue<int> queue;
  pthread_mutex_t queue_lock;
  pthread_cond_t  queue_cond;

public:
  Server(int port, int num, std::string root)
      : port(port), worker_num(num), doc_root(root) {
    pthread_mutex_init(&queue_lock, NULL);
    pthread_cond_init(&queue_cond, NULL);
  }
  void run() {
    const int soc = start_listening(this->port);
    logger::log("info", __FILE__, __LINE__, "listening...");
    prepare_workers();

    fd_set rfds;
    struct timeval timeout;
    while (g_stop_server == false) {
      FD_ZERO(&rfds);
      FD_SET(soc, &rfds);
      timeout.tv_sec = 0;
      timeout.tv_usec = 100 * 1000;
      int ret_sel = select(soc+1, &rfds, NULL, NULL, &timeout);
      if (ret_sel < 0) {
        if (errno == EINTR) {
          // シグナルキャッチ
        } else {
          logger::log("warn", __FILE__, __LINE__, "select error: %s", strerror(errno));
        }
      } else if (ret_sel == 0) {
        // timeout
        // logger::log("info", "timeout");
      } else if (FD_ISSET(soc, &rfds)) {
        sockaddr_in from;
        socklen_t len = sizeof(from);
        int acc = ::accept(soc, (sockaddr*)&from, &len);
        if (soc < 0) {
          logger::log("warn", __FILE__, __LINE__, "accept error: %s", strerror(errno));
          // retry
        } else {
          logger::log("info", __FILE__, __LINE__, "accept: %s:%d", inet_ntoa(from.sin_addr), ntohs(from.sin_port));
          pthread_mutex_lock(&(this->queue_lock));
          queue.push(acc);
          pthread_cond_broadcast(&(this->queue_cond));
          pthread_mutex_unlock(&(this->queue_lock));
        }
      }
    }
    // TODO woker stop
  }
private:
  int start_listening(const int port) {
    int soc = ::socket(AF_INET, SOCK_STREAM, 0);
    if (soc < 0) {
      SYS_ERROR();
    }
    int opt = 1;
    if (::setsockopt(soc, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, sizeof(int)) != 0) {
      SYS_ERROR();
    }
    struct sockaddr_in self;
    init_sockaddr_in(self, port);
    if (::bind(soc, (sockaddr*)&self, sizeof(self)) < 0) {  // success: 0, failure: -1
      SYS_ERROR();
    }
    if(::listen(soc, SOMAXCONN) < 0) {
      SYS_ERROR();
    }
    logger::log("info", __FILE__, __LINE__, "server ready (port=%d).", port);
    return soc;
  }
  void init_sockaddr_in(sockaddr_in& self, const int port) {
    ::memset(&self, 0, sizeof(self));
    self.sin_family = AF_INET;
    self.sin_addr.s_addr = htonl(INADDR_ANY);
    self.sin_port = htons(port);
  }
  void prepare_workers() {
    for (int i = 0; i < this->worker_num; ++i) {
      pthread_t id;
      pthread_create(&id, NULL, worker_thread, this);
      this->workers.push_back(id);
    }
  }
private:  // on worker thread
  const std::map<std::string, std::string> create_env_map() {
    std::map<std::string, std::string> map;
    map.insert(
        std::map<std::string, std::string>::value_type("DOC_ROOT", this->doc_root));
    return map;
  }
  static void* worker_thread(void* arg) {
    Server* server = reinterpret_cast<Server*>(arg);
    const std::map<std::string, std::string>& env = server->create_env_map();
    while (true) {
      pthread_mutex_lock(&(server->queue_lock));
      std::queue<int>& queue = server->queue;
      while (queue.empty()) {
        pthread_cond_wait(&(server->queue_cond), &(server->queue_lock));
      }
      const int acc = queue.front(); queue.pop();
      pthread_mutex_unlock(&(server->queue_lock));
      if (acc < 0) {  // 終了を期待されている？
        return NULL;
      }
      logger::log("info", __FILE__, __LINE__, "accept worker thread");
      do_work(acc, env);
    }
    return NULL;
  }
  static void do_work(const int acc, const std::map<std::string, std::string>& env) {
    http::run(acc, env);
  }
};

#include "options.hpp"
static void stop_server() {
  g_stop_server = true;
}
static void on_signal(int signum) {
  stop_server();
}
static void set_signal_handler(const int signum, void (*handler)(int)) {
  struct sigaction sa;
  memset(&sa, 0, sizeof(sa));
  sa.sa_handler = handler;
  sa.sa_flags = SA_RESTART;
  if (sigaction(signum, &sa, NULL) != 0) {
    SYS_ERROR();
  }
}
int main(int argc, char* argv[]) {
  Options opts(argc, argv);
  if (opts.has_error()) {
    std::cerr << opts.error() << std::endl;
    return 1;
  }

  // g_log = fopen(opts.log_path(), "a");
  if (opts.is_daemon()) {
    if (daemon(0, 0) == -1) {
      perror("daemon");
      return 1;
    }
  } else {
    logger::log("info", __FILE__, __LINE__, "options: %s", opts.to_string().c_str());
  }

  set_signal_handler(SIGINT, on_signal);
  set_signal_handler(SIGTERM, on_signal);
  set_signal_handler(SIGPIPE, SIG_IGN);

  Server server(opts.port(), opts.worker_num(), opts.doc_root());
  server.run();
  logger::log("info", __FILE__, __LINE__, "stop server.");
  //fclose(g_log);
  return 0;
}
